// Copyright (C) 2019 Storj Labs, Inc.
// See LICENSE for copying information.

package contact

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/zeebo/errs"
	"go.uber.org/zap"

	"storj.io/common/identity"
	"storj.io/common/pb"
	"storj.io/common/rpc/rpcstatus"
	"storj.io/common/storj"
	"storj.io/storj/private/nodeoperator"
	"storj.io/storj/satellite/overlay"
)

var (
	errPingBackDial     = errs.Class("pingback dialing")
	errCheckInIdentity  = errs.Class("check-in identity")
	errCheckInRateLimit = errs.Class("check-in ratelimit")
	errCheckInNetwork   = errs.Class("check-in network")
)

// Endpoint implements the contact service Endpoints.
type Endpoint struct {
	pb.DRPCNodeUnimplementedServer
	log     *zap.Logger
	service *Service
	disablePingBack bool

	sync.Mutex
	remoteAdressCache map[string]string
	pingResponeMessage string
}

// NewEndpoint returns a new contact service endpoint.
func NewEndpoint(log *zap.Logger, service *Service) *Endpoint {
	return &Endpoint{
		log:     log,
		service: service,
		disablePingBack: false,
		remoteAdressCache: make(map[string]string),
	}
}
func (endpoint *Endpoint)SetPingBack(dis bool){
	endpoint.disablePingBack = dis
}
func (endpoint *Endpoint)SetPingRespone(m string )  {
	endpoint.pingResponeMessage = m
}
// CheckIn is periodically called by storage nodes to keep the satellite informed of its existence,
// address, and operator information. In return, this satellite keeps the node informed of its
// reachability.
// When a node checks-in with the satellite, the satellite pings the node back to confirm they can
// successfully connect.
func (endpoint *Endpoint) CheckIn(ctx context.Context, req *pb.CheckInRequest, remoteAddr string) (_ *pb.CheckInResponse, err error) {
	defer mon.Task()(&ctx)(&err)
	peerID, err := identity.PeerIdentityFromContext(ctx)
	if err != nil {
		endpoint.log.Info("failed to get node ID from context", zap.String("node address", req.Address), zap.Error(err))
		return nil, rpcstatus.Error(rpcstatus.Unknown, errCheckInIdentity.New("failed to get ID from context: %v", err).Error())
	}
	nodeID := peerID.ID
	//fmt.Printf("%s checkin , %s remote address %s \n",req.P2PAddress, req.Address,  remoteAddr)
	endpoint.log.Debug("check in: ",zap.String("",req.P2PAddress),  zap.String("",remoteAddr))

	//save remoteAdrr
	nodeIdString := nodeID.String()
	endpoint.Lock()
	oldAddress := endpoint.remoteAdressCache[nodeIdString]
	if oldAddress != remoteAddr{

		endpoint.log.Debug(fmt.Sprintf("change remoteAddr %s --> %s \n",oldAddress, remoteAddr))
	}
	endpoint.remoteAdressCache[nodeIdString] = remoteAddr
	endpoint.Unlock()

	if endpoint.disablePingBack{
		return &pb.CheckInResponse{
			PingNodeSuccess:     true,
			PingNodeSuccessQuic: true,
			PingErrorMessage:    "",
		}, nil
	}
	//如果数据是空的，是快速的心中包，直接忽略掉
	if len(req.Address) == 0{
		return &pb.CheckInResponse{
			PingNodeSuccess:     true,
			PingNodeSuccessQuic: true,
			PingErrorMessage:    "",
		}, nil
	}

	// we need a string as a key for the limiter, but nodeID.String() has base58 encoding overhead
	nodeIDBytesAsString := string(nodeID.Bytes())
	if !endpoint.service.idLimiter.IsAllowed(nodeIDBytesAsString) {
		endpoint.log.Info("node rate limited by id", zap.String("node address", req.Address), zap.Stringer("Node ID", nodeID))
		return nil, rpcstatus.Error(rpcstatus.ResourceExhausted, errCheckInRateLimit.New("node rate limited by id").Error())
	}

	err = endpoint.service.peerIDs.Set(ctx, nodeID, peerID)
	if err != nil {
		endpoint.log.Info("failed to add peer identity entry for ID", zap.String("node address", req.Address), zap.Stringer("Node ID", nodeID), zap.Error(err))
		return nil, rpcstatus.Error(rpcstatus.FailedPrecondition, errCheckInIdentity.New("failed to add peer identity entry for ID: %v", err).Error())
	}

	//resolvedIPPort, resolvedNetwork, err := overlay.ResolveIPAndNetwork(ctx, req.Address)
	resolvedIPPort, resolvedNetwork, err := overlay.ResolveIPAndNetwork(ctx, remoteAddr)

	if err != nil {
		endpoint.log.Info("failed to resolve IP from address", zap.String("node address", req.Address), zap.Stringer("Node ID", nodeID), zap.Error(err))
		return nil, rpcstatus.Error(rpcstatus.InvalidArgument, errCheckInNetwork.New("failed to resolve IP from address: %s, err: %v", req.Address, err).Error())
	}

	//nodeurl := storj.NodeURL{
	//	ID:      nodeID,
	//	//Address: req.Address,
	//	Address: remoteAddr,
	//}
	//pingNodeSuccess, pingNodeSuccessQUIC, pingErrorMessage, err := endpoint.service.PingBack(ctx, nodeurl)
	pingNodeSuccess := true
	pingNodeSuccessQUIC := true
	pingErrorMessage := ""
	err = nil

	//if err != nil {
	//	endpoint.log.Info("failed to ping back address", zap.String("node address", req.Address), zap.Stringer("Node ID", nodeID), zap.Error(err))
	//	if errPingBackDial.Has(err) {
	//		err = errCheckInNetwork.New("failed dialing address when attempting to ping node (ID: %s): %s, err: %v", nodeID, req.Address, err)
	//		return nil, rpcstatus.Error(rpcstatus.NotFound, err.Error())
	//	}
	//	err = errCheckInNetwork.New("failed to ping node (ID: %s) at address: %s, err: %v", nodeID, req.Address, err)
	//	return nil, rpcstatus.Error(rpcstatus.NotFound, err.Error())
	//}

	// check wallet features
	if req.Operator != nil {
		if err := nodeoperator.DefaultWalletFeaturesValidation.Validate(req.Operator.WalletFeatures); err != nil {
			endpoint.log.Debug("ignoring invalid wallet features",
				zap.Stringer("Node ID", nodeID),
				zap.Strings("Wallet Features", req.Operator.WalletFeatures))

			// TODO: Update CheckInResponse to include wallet feature validation error
			req.Operator.WalletFeatures = nil
		}
	}

	nodeInfo := overlay.NodeCheckInInfo{
		NodeID: peerID.ID,
		Address: &pb.NodeAddress{
			Address:   remoteAddr,
			Transport: pb.NodeTransport_TCP_TLS_GRPC,
		},
		P2PAddress: req.P2PAddress,
		LastNet:    resolvedNetwork,
		LastIPPort: resolvedIPPort,
		IsUp:       pingNodeSuccess,
		Capacity:   req.Capacity,
		Operator:   req.Operator,
		Version:    req.Version,
	}

	err = endpoint.service.overlay.UpdateCheckIn(ctx, nodeInfo, time.Now().UTC())
	if err != nil {
		endpoint.log.Info("failed to update check in", zap.String("node address", req.Address), zap.Stringer("Node ID", nodeID), zap.Error(err))
		return nil, rpcstatus.Error(rpcstatus.Internal, Error.Wrap(err).Error())
	}

	endpoint.log.Debug("checking in", zap.Stringer("Node ID", nodeID), zap.String("node addr", req.Address),
		zap.String("node addr", req.P2PAddress), zap.Bool("ping node quic success", pingNodeSuccessQUIC),
		zap.String("ping node err msg", pingErrorMessage))
	return &pb.CheckInResponse{
		PingNodeSuccess:     pingNodeSuccess,
		PingNodeSuccessQuic: pingNodeSuccessQUIC,
		PingErrorMessage:    pingErrorMessage,
		UserMessage: 		 endpoint.pingResponeMessage,
	}, nil
}

// GetTime returns current timestamp.
func (endpoint *Endpoint) GetTime(ctx context.Context, req *pb.GetTimeRequest) (_ *pb.GetTimeResponse, err error) {
	defer mon.Task()(&ctx)(&err)

	peerID, err := identity.PeerIdentityFromContext(ctx)
	if err != nil {
		endpoint.log.Info("failed to get node ID from context", zap.Error(err))
		return nil, rpcstatus.Error(rpcstatus.Unauthenticated, errCheckInIdentity.New("failed to get ID from context: %v", err).Error())
	}

	currentTimestamp := time.Now().UTC()
	endpoint.log.Debug("get system current time", zap.Stringer("timestamp", currentTimestamp), zap.Stringer("node id", peerID.ID))
	return &pb.GetTimeResponse{
		Timestamp: currentTimestamp,
	}, nil
}

func (endpoint *Endpoint) GetRemoteAdress(node string)string{
	endpoint.Lock()
	defer  endpoint.Unlock()
	//endpoint.log.Info("remoteAdressCache len ", zap.Int("", len(endpoint.remoteAdressCache)))
	 adress , ok := endpoint.remoteAdressCache[node]
	 if !ok{
	 	return ""
	 }
	 return adress
}

func (endpoint *Endpoint) NotifyPingGayway(ctx context.Context, nodeId storj.NodeID ,gatway string ){
	hole :=  endpoint.GetRemoteAdress(nodeId.String());
	if len(hole) == 0{
		endpoint.log.Warn("hole is empty",zap.String("",nodeId.String()))
		return
	}
	nodeurl := storj.NodeURL{
		//Address: req.Address,
		ID: nodeId,
		Address: hole,
	}
	endpoint.service.NotifyGateway(ctx, nodeurl,gatway)


}